package com.ljy.kotlindemo

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.system.measureTimeMillis

/**
 * @Author:         LiuJinYang
 * @CreateDate:     2020/10/19
 *
 * 协程(Coroutines)，是 Kotlin 最神奇的特性
 */
fun main() {
//    testCoroutineBase()
//    testCancelAndTimeout()
//    testAsync()
//    testDispatcher()
    testAsyncStream()
}

//一. 协程基础
fun testCoroutineBase() {
    //1. 第一个协程程序
    GlobalScope.launch { // 在后台启动一个新的协程并继续
        delay(1000L) // 非阻塞的等待 1 秒钟（默认时间单位是毫秒）
        // delay 是一个特殊的 挂起函数 ，它不会造成线程阻塞，但是会 挂起 协程，并且只能在协程中使用
        println("World!") // 在延迟后打印输出
    }
    println("Hello,") // 协程已在等待时主线程还在继续
    Thread.sleep(2000L) // 阻塞主线程 2 秒钟来保证 JVM 存活
    //本质上，协程是轻量级的线程。 它们在某些 CoroutineScope 上下文中与 launch 协程构建器 一起启动。
    // 这里我们在 GlobalScope 中启动了一个新的协程，这意味着新协程的生命周期只受整个应用程序的生命周期限制。
    //2. 桥接阻塞与非阻塞的世界
    //非阻塞的 delay(……) 与 阻塞的 Thread.sleep(……)。 这容易让我们记混哪个是阻塞的、哪个是非阻塞的。
    // 让我们显式使用 runBlocking 协程构建器来阻塞：
    GlobalScope.launch { // 在后台启动一个新的协程并继续
        delay(1000L)
        println("World22222!")
    }
    println("Hello22222,") // 主线程中的代码会立即执行
    runBlocking {     // 但是这个表达式阻塞了主线程
        delay(2000L)  // ……我们延迟 2 秒来保证 JVM 的存活
    }
    //调用了 runBlocking 的主线程会一直 阻塞 直到 runBlocking 内部的协程执行完毕
    //这个示例可以使用更合乎惯用法的方式重写
    runBlocking<Unit> { // 开始执行主协程
        GlobalScope.launch { // 在后台启动一个新的协程并继续
            delay(1000L)
            println("World3333!")
        }
        println("Hello3333,") // 主协程在这里会立即执行
        delay(2000L)      // 延迟 2 秒来保证 JVM 存活
    }
    //3. 等待一个作业
    //延迟一段时间来等待另一个协程运行并不是一个好的选择。让我们显式（以非阻塞方式）等待所启动的后台 Job 执行结束：
    runBlocking {
        val job = GlobalScope.launch { // 启动一个新协程并保持对这个作业的引用
            delay(1000L)
            println("World4444!")
        }
        println("Hello4444,")
        job.join() // 等待直到子协程执行结束
    }
    //4. 结构化的并发
    runBlocking { // this: CoroutineScope
        launch { // 在 runBlocking 作用域中启动一个新协程
            delay(1000L)
            println("World5555!")
        }
        println("Hello5555,")
    }
    //4. 作用域构建器
    //除了由不同的构建器提供协程作用域之外，还可以使用 coroutineScope 构建器声明自己的作用域。
    // 它会创建一个协程作用域并且在所有已启动子协程执行完毕之前不会结束。
    //runBlocking 与 coroutineScope 可能看起来很类似，因为它们都会等待其协程体以及所有子协程结束。
    // 主要区别在于，runBlocking 方法会阻塞当前线程来等待， 而 coroutineScope 只是挂起，会释放底层线程用于其他用途。
    // 由于存在这点差异，runBlocking 是常规函数，而 coroutineScope 是挂起函数
    runBlocking { // this: CoroutineScope
        launch {
            delay(200L)
            println("Task from runBlocking")
        }

        coroutineScope { // 创建一个协程作用域
            launch {
                delay(500L)
                println("Task from nested launch")
            }

            delay(100L)
            println("Task from coroutine scope") // 这一行会在内嵌 launch 之前输出
        }

        println("Coroutine scope is over") // 这一行在内嵌 launch 执行完毕后才输出
    }

    //5. 提取函数重构
    //我们来将 launch { …… } 内部的代码块提取到独立的函数中。当你对这段代码执行“提取函数”重构时，
    // 你会得到一个带有 suspend 修饰符的新函数。 这是你的第一个挂起函数。在协程内部可以像普通函数一样使用挂起函数，
    // 不过其额外特性是，同样可以使用其他挂起函数（如本例中的 delay）来挂起协程的执行
    runBlocking {
        launch { doWorld() }
        println("Hello,")
    }
    //6. 协程很轻量
    //它启动了 10 万个协程，并且在 5 秒钟后，每个协程都输出一个点。
    //尝试使用线程来实现。会发生什么？（很可能你的代码会产生某种内存不足的错误）
    runBlocking {
        repeat(100_000) { // 启动大量的协程
            launch {
                delay(5000L)
                print(".")
            }
        }
    }
    //7. 全局协程像守护线程
    runBlocking {
        GlobalScope.launch {
            repeat(1000) { i ->
                println("I'm sleeping $i ...")
                delay(500L)
            }
        }
        delay(1300L) // 在延迟后退出
    }

}

// 这是你的第一个挂起函数
suspend fun doWorld() {
    delay(1000L)
    println("World!")
}

//二。取消与超时
fun testCancelAndTimeout() {
    //1. 取消协程的执行
    //在一个长时间运行的应用程序中，你也许需要对你的后台协程进行细粒度的控制。
    // 比如说，一个用户也许关闭了一个启动了协程的界面，那么现在协程的执行结果已经不再被需要了，这时，它应该是可以被取消的。
    // 该 launch 函数返回了一个可以被用来取消运行中的协程的 Job
    runBlocking {
        val job = launch {
            repeat(1000) { i ->
                println("job: I'm sleeping $i ...")
                delay(500L)
            }
        }
        delay(1300L) // 延迟一段时间
        println("main: I'm tired of waiting!")
        job.cancel() // 取消该作业
        job.join() // 等待作业执行结束
        println("main: Now I can quit.")
    }
    //2. 取消是协作的
    //协程的取消是 协作 的。一段协程代码必须协作才能被取消。
    // 所有 kotlinx.coroutines 中的挂起函数都是 可被取消的 。
    // 它们检查协程的取消， 并在取消时抛出 CancellationException。
    // 然而，如果协程正在执行计算任务，并且没有检查取消的话，那么它是不能被取消的，就如如下示例代码所示：
    runBlocking {
        val startTime = System.currentTimeMillis()
        val job = launch(Dispatchers.Default) {
            var nextPrintTime = startTime
            var i = 0
            while (i < 5) { // 一个执行计算的循环，只是为了占用 CPU
                // 每秒打印消息两次
                if (System.currentTimeMillis() >= nextPrintTime) {
                    println("job: I'm sleeping ${i++} ...")
                    nextPrintTime += 500L
                }
            }
        }
        delay(1300L) // 等待一段时间
        println("main: I'm tired of waiting!")
        job.cancelAndJoin() // 取消一个作业并且等待它结束
        println("main: Now I can quit.")
    }
    //3. 使计算代码可取消
    //有两种方法来使执行计算的代码可以被取消。第一种方法是定期调用挂起函数来检查取消。
    // 对于这种目的 yield 是一个好的选择。 另一种方法是显式的检查取消状态。让我们试试第二种方法。
    runBlocking {
        val startTime = System.currentTimeMillis()
        val job = launch(Dispatchers.Default) {
            var nextPrintTime = startTime
            var i = 0
            while (isActive) { // 可以被取消的计算循环
                // 每秒打印消息两次
                if (System.currentTimeMillis() >= nextPrintTime) {
                    println("job: I'm sleeping ${i++} ...")
                    nextPrintTime += 500L
                }
            }
        }
        delay(1300L) // 等待一段时间
        println("main: I'm tired of waiting!")
        job.cancelAndJoin() // 取消该作业并等待它结束
        println("main: Now I can quit.")
    }
    //4. 在 finally 中释放资源
    //我们通常使用如下的方法处理在被取消时抛出 CancellationException 的可被取消的挂起函数。
    // 比如说，try {……} finally {……} 表达式以及 Kotlin 的 use 函数一般在协程被取消的时候执行它们的终结动作：
    runBlocking {
        val job = launch {
            try {
                repeat(1000) { i ->
                    println("job: I'm sleeping $i ...")
                    delay(500L)
                }
            } finally {
                println("job: I'm running finally")
            }
        }
        delay(1300L) // 延迟一段时间
        println("main: I'm tired of waiting!")
        job.cancelAndJoin() // 取消该作业并且等待它结束
        println("main: Now I can quit.")
    }
    //5. 运行不能取消的代码块
    //在前一个例子中任何尝试在 finally 块中调用挂起函数的行为都会抛出 CancellationException，
    // 因为这里持续运行的代码是可以被取消的。通常，这并不是一个问题，所有良好的关闭操作（关闭一个文件、取消一个作业、
    // 或是关闭任何一种通信通道）通常都是非阻塞的，并且不会调用任何挂起函数。然而，在真实的案例中，当你需要挂起
    // 一个被取消的协程，你可以将相应的代码包装在 withContext(NonCancellable) {……} 中，并使用 withContext
    // 函数以及 NonCancellable 上下文，见如下示例所示：
    runBlocking {
        val job = launch {
            try {
                repeat(1000) { i ->
                    println("job: I'm sleeping $i ...")
                    delay(500L)
                }
            } finally {
                withContext(NonCancellable) {
                    println("job: I'm running finally")
                    delay(1000L)
                    println("job: And I've just delayed for 1 sec because I'm non-cancellable")
                }
            }
        }
        delay(1300L) // 延迟一段时间
        println("main: I'm tired of waiting!")
        job.cancelAndJoin() // 取消该作业并等待它结束
        println("main: Now I can quit.")
    }
    //6. 超时
    //在实践中绝大多数取消一个协程的理由是它有可能超时。 当你手动追踪一个相关 Job 的引用并启动了一个单独的协程
    // 在延迟后取消追踪，这里已经准备好使用 withTimeout 函数来做这件事。 来看看示例代码：
//    runBlocking {
//        withTimeout(1300L) {
//            repeat(1000) { i ->
//                println("I'm sleeping $i ...")
//                delay(500L)
//            }
//        }
//    }
    // 如果你需要做一些各类使用超时的特别的额外操作，可以使用类似 withTimeout 的 withTimeoutOrNull 函数，
    // 并把这些会超时的代码包装在 try {...} catch (e: TimeoutCancellationException) {...} 代码块中，
    // 而 withTimeoutOrNull 通过返回 null 来进行超时操作，从而替代抛出一个异常：
    runBlocking {
        val result = withTimeoutOrNull(1300L) {
            repeat(1000) { i ->
                println("I'm sleeping $i ...")
                delay(500L)
            }
            "Done" // 在它运行得到结果之前取消它
        }
        println("Result is $result")
    }
    //7。异步超时和资源
    //对于在其块中运行的代码，withTimeout中的timeout事件是异步的，并且可能在任何时间发生，
    // 甚至就在从timeout块内部返回之前。如果您在块内打开或获取一些需要关闭或在块外释放的资源，请记住这一点。
    //例如，在这里，我们使用Resource该类来模仿一个可关闭的资源，该类通过递增acquired计数器并使该计数器从其close函数
    // 递减来简单地跟踪它被创建了多少次。让我们以少量超时运行大量协程，尝试withTimeout在延迟一段时间后从块内部获取此资源，然后从外部释放它
    var acquired = 0

    class Resource {
        init {
            acquired++
        } // Acquire the resource

        fun close() {
            acquired--
        } // Release the resource
    }
    runBlocking {
        acquired = 0
        repeat(100_000) { // Launch 100K coroutines
            launch {
                val resource = withTimeout(60) { // Timeout of 60 ms
                    delay(50) // Delay for 50 ms
                    Resource() // Acquire a resource and return it from withTimeout block
                }
                resource.close() // Release the resource
            }
        }
    }
    // Outside of runBlocking all coroutines have completed
    println(acquired) // Print the number of resources still acquired
    //如果运行上面的代码，您会发现它并不总是输出零，尽管它可能取决于计算机的时序，但在此示例中，您可能需要调整超时才能实际看到非零值。
    //请注意，acquired此处从100K协程递增和递减计数器是完全安全的，因为它总是发生在同一主线程中。关于这一点的更多内容将在下一章有关协程的上下文中进行解释。
    //要解决此问题，您可以将对资源的引用存储在变量中，而不是从withTimeout块中返回资源。
    //本示例始终打印零。资源不会泄漏。
    runBlocking {
        acquired = 0
        repeat(100_000) { // 启动100k个协程
            launch {
                var resource: Resource? = null // 尚未获取
                try {
                    withTimeout(60) { // 超时时间为60毫秒
                        delay(50) // 延迟50毫秒
                        resource = Resource() // 将资源存储到变量（如果已获取）
                    }
                    //我们可以使用此处的资源做其他事情
                } finally {
                    resource?.close() // 释放已获取资源
                }
            }
        }
    }
// 在runBlocking之外，所欲呕协同程序都已完成
    println(acquired) // 打印获得的资源数量
}

//三。组合挂起函数
fun testAsync() {
    //1. 默认顺序调用
    //假设我们在不同的地方定义了两个进行某种调用远程服务或者进行计算的挂起函数。
    // 我们只假设它们都是有用的，但是实际上它们在这个示例中只是为了该目的而延迟了一秒钟：
    suspend fun doSomethingUsefulOne(): Int {
        delay(3000L) // 假设我们在这里做了一些有用的事
        println("doSomethingUsefulOne")
        return 13
    }

    suspend fun doSomethingUsefulTwo(): Int {
        delay(1000L) // 假设我们在这里也做了一些有用的事
        println("doSomethingUsefulTwo")
        return 29
    }
    //如果需要按 顺序 调用它们，我们接下来会做什么？首先调用 doSomethingUsefulOne ，
    // 接下来 调用 doSomethingUsefulTwo，并且计算它们结果的和吗？ 实际上，
    // 如果我们要根据第一个函数的结果来决定是否我们需要调用第二个函数或者决定如何调用它时，我们就会这样做。
    //我们使用普通的顺序来进行调用，因为这些代码是运行在协程中的，只要像常规的代码一样 顺序 都是默认的。
    // 下面的示例展示了测量执行两个挂起函数所需要的总时间：
    runBlocking {
        val time = measureTimeMillis {
            val one = doSomethingUsefulOne()
            val two = doSomethingUsefulTwo()
            println("11111The answer is ${one + two}")
        }
        println("11111Completed in $time ms")
    }
    //2. 使用 async 并发
    //如果 doSomethingUsefulOne 与 doSomethingUsefulTwo 之间没有依赖，并且我们想更快的得到结果，
    // 让它们进行 并发 吗？这就是 async 可以帮助我们的地方。
    //在概念上，async 就类似于 launch。它启动了一个单独的协程，这是一个轻量级的线程并与其它所有的协程一起并发的工作。
    // 不同之处在于 launch 返回一个 Job 并且不附带任何结果值，而 async 返回一个 Deferred —— 一个轻量级的非阻塞 future，
    // 这代表了一个将会在稍后提供结果的 promise。你可以使用 .await() 在一个延期的值上得到它的最终结果，
    // 但是 Deferred 也是一个 Job，所以如果需要的话，你可以取消它。
    runBlocking {
        val time = measureTimeMillis {
            val one = async { doSomethingUsefulOne() }
            val two = async { doSomethingUsefulTwo() }
            println("22222The answer is ${one.await() + two.await()}")
        }
        println("22222Completed in $time ms")
    }
    //3. 惰性启动的 async
    //可选的，async 可以通过将 start 参数设置为 CoroutineStart.LAZY 而变为惰性的。 在这个模式下，
    // 只有结果通过 await 获取的时候协程才会启动，或者在 Job 的 start 函数调用的时候
    //注意，如果我们只是在 println 中调用 await，而没有在单独的协程中调用 start，这将会导致顺序行为，
    // 直到 await 启动该协程 执行并等待至它结束，这并不是惰性的预期用例
    runBlocking<Unit> {
        val time = measureTimeMillis {
            val one = async(start = CoroutineStart.LAZY) { doSomethingUsefulOne() }
            val two = async(start = CoroutineStart.LAZY) { doSomethingUsefulTwo() }
            // 执行一些计算
            one.start() // 启动第一个
            two.start() // 启动第二个
            println("33333The answer is ${one.await() + two.await()}")
        }
        println("33333Completed in $time ms")
    }
    //4. async 风格的函数
    //我们可以定义异步风格的函数来 异步 的调用 doSomethingUsefulOne 和 doSomethingUsefulTwo
    // 并使用 async 协程建造器并带有一个显式的 GlobalScope 引用。
    // 我们给这样的函数的名称中加上“……Async”后缀来突出表明：事实上，它们只做异步计算并且需要使用延期的值来获得结果
    // somethingUsefulOneAsync 函数的返回值类型是 Deferred<Int>
    fun somethingUsefulOneAsync() = GlobalScope.async {
        doSomethingUsefulOne()
    }

    // somethingUsefulTwoAsync 函数的返回值类型是 Deferred<Int>
    fun somethingUsefulTwoAsync() = GlobalScope.async {
        doSomethingUsefulTwo()
    }
    //注意，这些 xxxAsync 函数不是 挂起 函数。它们可以在任何地方使用。
    // 然而，它们总是在调用它们的代码中意味着异步（这里的意思是 并发 ）执行。
    //下面的例子展示了它们在协程的外面是如何使用的：
    val time = measureTimeMillis {
        // 我们可以在协程外面启动异步执行
        val one = somethingUsefulOneAsync()
        val two = somethingUsefulTwoAsync()
        // 但是等待结果必须调用其它的挂起或者阻塞
        // 当我们等待结果的时候，这里我们使用 `runBlocking { …… }` 来阻塞主线程
        runBlocking {
            println("44444The answer is ${one.await() + two.await()}")
        }
    }
    println("44444Completed in $time ms")
    //这种带有异步函数的编程风格仅供参考，因为这在其它编程语言中是一种受欢迎的风格。在 Kotlin 的协程中使用这种风格是强烈不推荐的，
    // 原因如下所述。
    //考虑一下如果 val one = somethingUsefulOneAsync() 这一行和 one.await() 表达式这里在代码中有逻辑错误，
    // 并且程序抛出了异常以及程序在操作的过程中中止，将会发生什么。 通常情况下，一个全局的异常处理者会捕获这个异常，
    // 将异常打印成日记并报告给开发者，但是反之该程序将会继续执行其它操作。但是这里我们的 somethingUsefulOneAsync
    // 仍然在后台执行， 尽管如此，启动它的那次操作也会被终止。这个程序将不会进行结构化并发，如下一小节所示。
    //5. 使用 async 的结构化并发
    //让我们使用使用 async 的并发这一小节的例子并且提取出一个函数并发的调用 doSomethingUsefulOne 与 doSomethingUsefulTwo
    // 并且返回它们两个的结果之和。 由于 async 被定义为了 CoroutineScope 上的扩展，我们需要将它写在作用域内，并且这是 coroutineScope 函数所提供的：
    suspend fun concurrentSum(): Int = coroutineScope {
        val one = async { doSomethingUsefulOne() }
        val two = async { doSomethingUsefulTwo() }
        one.await() + two.await()
    }
    //这种情况下，如果在 concurrentSum 函数内部发生了错误，并且它抛出了一个异常， 所有在作用域中启动的协程都会被取消。
    runBlocking {
        val time = measureTimeMillis {
            println("The answer is ${concurrentSum()}")
        }
        println("Completed in $time ms")
    }
    //取消始终通过协程的层次结构来进行传递：
    suspend fun failedConcurrentSum(): Int = coroutineScope {
        val one = async<Int> {
            try {
                delay(Long.MAX_VALUE) // 模拟一个长时间的运算
                42
            } finally {
                println("First child was cancelled")
            }
        }
        val two = async<Int> {
            println("Second child throws an exception")
            throw ArithmeticException()
        }
        one.await() + two.await()
    }
    //请注意，如果其中一个子协程（即 two）失败，第一个 async 以及等待中的父协程都会被取消：
    runBlocking<Unit> {
        try {
            failedConcurrentSum()
        } catch (e: ArithmeticException) {
            println("Computation failed with ArithmeticException")
        }
    }
}

//四。协程上下文与调度器
fun testDispatcher() {
    //协程总是运行在一些以 CoroutineContext 类型为代表的上下文中，它们被定义在了 Kotlin 的标准库里
    //1. 调度器与线程
    //协程上下文包含一个 协程调度器 （参见 CoroutineDispatcher）它确定了相关的协程在哪个线程或哪些线程上执行。
    // 协程调度器可以将协程限制在一个特定的线程执行，或将它分派到一个线程池，亦或是让它不受限地运行。
    //所有的协程构建器诸如 launch 和 async 接收一个可选的 CoroutineContext 参数，它可以被用来显式的为一个新协程或其它上下文元素指定一个调度器。
    runBlocking<Unit> {
        launch { // 运行在父协程的上下文中，即 runBlocking 主协程
            println("main runBlocking      : I'm working in thread ${Thread.currentThread().name}")
        }
        launch(Dispatchers.Unconfined) { // 不受限的——将工作在主线程中
            println("Unconfined            : I'm working in thread ${Thread.currentThread().name}")
        }
        launch(Dispatchers.Default) { // 将会获取默认调度器
            //launch(Dispatchers.Default) { …… } 与 GlobalScope.launch { …… } 使用相同的调度器。
            println("Default               : I'm working in thread ${Thread.currentThread().name}")
        }
        launch(newSingleThreadContext("MyOwnThread")) { // 将使它获得一个新的线程
            //newSingleThreadContext 为协程的运行启动了一个线程。 一个专用的线程是一种非常昂贵的资源。
            // 在真实的应用程序中两者都必须被释放，当不再需要的时候，使用 close 函数，或存储在一个顶层变量中使它在整个应用程序中被重用
            println("newSingleThreadContext: I'm working in thread ${Thread.currentThread().name}")
        }
    }

    //2. 非受限调度器 vs 受限调度器
    //Dispatchers.Unconfined 协程调度器在调用它的线程启动了一个协程，但它仅仅只是运行到第一个挂起点。挂起后，
    // 它恢复线程中的协程，而这完全由被调用的挂起函数来决定。非受限的调度器非常适用于执行不消耗 CPU 时间的任务，
    // 以及不更新局限于特定线程的任何共享数据（如UI）的协程。
    //另一方面，该调度器默认继承了外部的 CoroutineScope。 runBlocking 协程的默认调度器，特别是， 当它被限制在了调用者线程时，
    // 继承自它将会有效地限制协程在该线程运行并且具有可预测的 FIFO 调度。
    runBlocking<Unit> {
        launch(Dispatchers.Unconfined) { // 非受限的——将和主线程一起工作
            println("Unconfined      : I'm working in thread ${Thread.currentThread().name}")
            delay(500)
            println("Unconfined      : After delay in thread ${Thread.currentThread().name}")
        }
        launch { // 父协程的上下文，主 runBlocking 协程
            println("main runBlocking: I'm working in thread ${Thread.currentThread().name}")
            delay(1000)
            println("main runBlocking: After delay in thread ${Thread.currentThread().name}")
        }
    }
    //所以，该协程的上下文继承自 runBlocking {...} 协程并在 main 线程中运行，当 delay 函数调用的时候，非受限的那个协程在默认的执行者线程中恢复执行。
    //非受限的调度器是一种高级机制，可以在某些极端情况下提供帮助而不需要调度协程以便稍后执行或产生不希望的副作用，
    // 因为某些操作必须立即在协程中执行。 非受限调度器不应该在通常的代码中使用。

    //3. 调试协程与线程
    //协程可以在一个线程上挂起并在其它线程上恢复。 如果没有特殊工具，甚至对于一个单线程的调度器也是难以弄清楚协程在何时何地正在做什么事情。
    //3.1:用 IDEA 调试
    //调试工具窗口包含 Coroutines 标签。在这个标签中，你可以同时找到运行中与已挂起的协程的相关信息。 这些协程以它们所运行的调度器进行分组
    //3.2:用日志调试
    //另一种调试线程应用程序而不使用协程调试器的方法是让线程在每一个日志文件的日志声明中打印线程的名字。
    // 这种特性在日志框架中是普遍受支持的。但是在使用协程时，单独的线程名称不会给出很多协程上下文信息，
    // 所以 kotlinx.coroutines 包含了调试工具来让它更简单。 使用 -Dkotlinx.coroutines.debug JVM 参数运行下面的代码：

    runBlocking<Unit> {
        val a = async {
            log("I'm computing a piece of the answer")
            6
        }
        val b = async {
            log("I'm computing another piece of the answer")
            7
        }
        log("The answer is ${a.await() * b.await()}")
    }

    //4. 在不同线程间跳转
//    newSingleThreadContext("Ctx1").use { ctx1 ->
//        newSingleThreadContext("Ctx2").use { ctx2 ->
//            runBlocking(ctx1) {
//                log("Started in ctx1")
//                withContext(ctx2) {
//                    log("Working in ctx2")
//                }
//                log("Back to ctx1")
//            }
//        }
//    }

    //5. 上下文中的作业
    //协程的 Job 是上下文的一部分，并且可以使用 coroutineContext [Job] 表达式在上下文中检索它：
    runBlocking<Unit> {
        println("My job is ${coroutineContext[Job]}")
    }

    //6. 子协程
    //当一个协程被其它协程在 CoroutineScope 中启动的时候， 它将通过 CoroutineScope.coroutineContext 来承袭上下文，
    // 并且这个新协程的 Job 将会成为父协程作业的 子 作业。当一个父协程被取消的时候，所有它的子协程也会被递归的取消。
    //然而，当使用 GlobalScope 来启动一个协程时，则新协程的作业没有父作业。 因此它与这个启动的作用域无关且独立运作。
    runBlocking<Unit> {
        // 启动一个协程来处理某种传入请求（request）
        val request = launch {
            // 孵化了两个子作业, 其中一个通过 GlobalScope 启动
            GlobalScope.launch {
                println("job1: I run in GlobalScope and execute independently!")
                delay(1000)
                println("job1: I am not affected by cancellation of the request")
            }
            // 另一个则承袭了父协程的上下文
            launch {
                delay(100)
                println("job2: I am a child of the request coroutine")
                delay(1000)
                println("job2: I will not execute this line if my parent request is cancelled")
            }
        }
        delay(500)
        request.cancel() // 取消请求（request）的执行
        delay(1000) // 延迟一秒钟来看看发生了什么
        println("main: Who has survived request cancellation?")
    }

    //7. 父协程的职责
    //一个父协程总是等待所有的子协程执行结束。父协程并不显式的跟踪所有子协程的启动，并且不必使用 Job.join 在最后的时候等待它们
    runBlocking<Unit> {
        // 启动一个协程来处理某种传入请求（request）
        val request = launch {
            repeat(5) { i -> // 启动少量的子作业
                launch {
                    delay((i + 1) * 200L) // 延迟 200 毫秒、400 毫秒、600 毫秒的时间
                    println("Coroutine $i is done")
                }
            }
            println("request: I'm done and I don't explicitly join my children that are still active")
        }
        request.join() // 等待请求的完成，包括其所有子协程
        println("Now processing of the request is complete")
    }

    //8.命名协程以用于调试
    //当协程经常打印日志并且你只需要关联来自同一个协程的日志记录时， 则自动分配的 id 是非常好的。
    // 然而，当一个协程与特定请求的处理相关联时或做一些特定的后台任务，最好将其明确命名以用于调试目的。
    // CoroutineName 上下文元素与线程名具有相同的目的。当调试模式开启时，它被包含在正在执行此协程的线程名中。
    runBlocking(CoroutineName("main")) {
        log("Started main coroutine")
        // 运行两个后台值计算
        val v1 = async(CoroutineName("v1coroutine")) {
            delay(500)
            log("Computing v1")
            252
        }
        val v2 = async(CoroutineName("v2coroutine")) {
            delay(1000)
            log("Computing v2")
            6
        }
        log("The answer for v1 / v2 = ${v1.await() / v2.await()}")
    }

    //9.组合上下文中的元素
    //有时我们需要在协程上下文中定义多个元素。我们可以使用 + 操作符来实现。
    // 比如说，我们可以显式指定一个调度器来启动协程并且同时显式指定一个命名：
    runBlocking<Unit> {
        launch(Dispatchers.Default + CoroutineName("test")) {
            println("I'm working in thread ${Thread.currentThread().name}")
        }
    }
    //JVM配置项-Dkotlinx.coroutines.debug:Run | Edit Configurations 点击进入粘贴配置到VM options

    //10. 协程作用域
    //让我们将关于上下文，子协程以及作业的知识综合在一起。假设我们的应用程序拥有一个具有生命周期的对象，
    // 但这个对象并不是一个协程。举例来说，我们编写了一个 Android 应用程序并在 Android 的 activity 上下文中
    // 启动了一组协程来使用异步操作拉取并更新数据以及执行动画等等。所有这些协程必须在这个 activity 销毁的时候
    // 取消以避免内存泄漏。当然，我们也可以手动操作上下文与作业，以结合 activity 的生命周期与它的协程，
    // 但是 kotlinx.coroutines 提供了一个封装：CoroutineScope 的抽象。 你应该已经熟悉了协程作用域，
    // 因为所有的协程构建器都声明为在它之上的扩展。
    //
    //我们通过创建一个 CoroutineScope 实例来管理协程的生命周期，并使它与 activity 的生命周期相关联。
    // CoroutineScope 可以通过 CoroutineScope() 创建或者通过MainScope() 工厂函数。前者创建了一个通用作用域，
    // 而后者为使用 Dispatchers.Main 作为默认调度器的 UI 应用程序 创建作用域：
    class Activity {
        private val mainScope = CoroutineScope(Dispatchers.Default) // use Default for test purposes
//        private val mainScope = MainScope()

        fun destroy() {
            mainScope.cancel()
        }

        fun doSomething() {
            // launch ten coroutines for a demo, each working for a different time
            repeat(10) { i ->
                mainScope.launch {
                    delay((i + 1) * 200L) // variable delay 200ms, 400ms, ... etc
                    println("Coroutine $i is done")
                }
            }
        }
    } // class Activity ends
    runBlocking<Unit> {
        val activity = Activity()
        activity.doSomething() // run test function
        println("Launched coroutines")
        delay(500L) // delay for half a second
        println("Destroying activity!")
        activity.destroy() // cancels all coroutines
        delay(1000) // visually confirm that they don't work
    }

    //11. 线程局部数据
    //有时，能够将一些线程局部数据传递到协程与协程之间是很方便的。 然而，由于它们不受任何特定线程的约束，如果手动完成，可能会导致出现样板代码。
    //ThreadLocal， asContextElement 扩展函数在这里会充当救兵。它创建了额外的上下文元素， 且保留给定 ThreadLocal 的值，
    // 并在每次协程切换其上下文时恢复它。 它很容易在下面的代码中演示：
    runBlocking<Unit> {
        threadLocal.set("main")
        println("Pre-main, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
        val job = launch(Dispatchers.Default + threadLocal.asContextElement(value = "launch")) {
            println("Launch start, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
            yield()
            println("After yield, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
        }
        job.join()
        println("Post-main, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
    }
    //在这个例子中我们使用 Dispatchers.Default 在后台线程池中启动了一个新的协程，所以它工作在线程池中的不同线程中，
    // 但它仍然具有线程局部变量的值， 我们指定使用 threadLocal.asContextElement(value = "launch")， 无论协程执行在哪个线程中都是没有问题的。

}

fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")

val threadLocal = ThreadLocal<String?>() // 声明线程局部变量

//五。异步流

fun testAsyncStream() {
    //挂起函数可以异步的返回单个值，但是该如何异步返回多个计算好的值呢？这正是 Kotlin 流（Flow）的用武之地。
    //1. 表示多个值
    //在 Kotlin 中可以使用集合来表示多个值。 比如说，我们有一个 simple 函数，它返回一个包含三个数字的 List， 然后使用 forEach 打印它们：
    fun simple1(): List<Int> = listOf(1, 2, 3)
    simple1().forEach { value -> println(value) }
    //1.1 序列
    //如果使用一些消耗 CPU 资源的阻塞代码计算数字（每次计算需要 1000 毫秒）那么我们可以使用 Sequence 来表示数字：
    fun simple2(): Sequence<Int> = sequence { // 序列构建器
        for (i in 1..3) {
            Thread.sleep(1000) // 假装我们正在计算
            yield(i) // 产生下一个值
        }
    }
    simple2().forEach { value -> println(value) }
    //1.2 挂起函数
    //然而，计算过程阻塞运行该代码的主线程。 当这些值由异步代码计算时，我们可以使用 suspend 修饰符标记函数 simple，
    // 这样它就可以在不阻塞的情况下执行其工作并将结果作为列表返回：
    suspend fun simple3(): List<Int> {
        delay(3000) // 假装我们在这里做了一些异步的事情
        return listOf(1, 2, 3)
    }
    runBlocking {
        simple3().forEach { value -> println(value) }
    }
    //1.3 流
    //使用 List 结果类型，意味着我们只能一次返回所有值。 为了表示异步计算的值流（stream），
    // 我们可以使用 Flow 类型（正如同步计算值会使用 Sequence 类型）：
    fun simple4(): Flow<Int> = flow { // 流构建器
        for (i in 1..3) {
            delay(1000) // 假装我们在这里做了一些有用的事情
            emit(i) // 发送下一个值
        }
    }
    runBlocking<Unit> {
        // 启动并发的协程以验证主线程并未阻塞
        launch {
            for (k in 1..3) {
                println("I'm not blocked $k")
                delay(1000)
            }
        }
        // 收集这个流
        simple4().collect { value -> println(value) }
    }

    //2. 流是冷的
    //Flow 是一种类似于序列的冷流 — 这段 flow 构建器中的代码直到流被收集的时候才运行。这在以下的示例中非常明显：
    fun simple5(): Flow<Int> = flow {
        println("Flow started")
        for (i in 1..3) {
            delay(100)
            emit(i)
        }
    }
    runBlocking<Unit> {
        println("Calling simple function...")
        val flow = simple5()
        println("Calling collect...")
        flow.collect { value -> println(value) }
        println("Calling collect again...")
        flow.collect { value -> println(value) }
    }
    //这是返回一个流的 simple 函数没有标记 suspend 修饰符的主要原因。 通过它自己，simple() 调用会尽快返回且不会进行任何等待。
    // 该流在每次收集的时候启动， 这就是为什么当我们再次调用 collect 时我们会看到“Flow started”
    //3. 流取消基础
    //流采用与协程同样的协作取消。像往常一样，流的收集可以在当流在一个可取消的挂起函数（例如 delay）中挂起的时候取消。
    fun simple6(): Flow<Int> = flow {
        for (i in 1..3) {
            delay(100)
            println("Emitting $i")
            emit(i)
        }
    }
    runBlocking<Unit> {
        withTimeoutOrNull(250) { // 在 250 毫秒后超时
            simple6().collect { value -> println(value) }
        }
        println("Done")
    }
    //4. 流构建器
    //先前示例中的 flow { ... } 构建器是最基础的一个。还有其他构建器使流的声明更简单：
    //a. flowOf 构建器定义了一个发射固定值集的流。
    //b. 使用 .asFlow() 扩展函数，可以将各种集合与序列转换为流。
    //因此，从流中打印从 1 到 3 的数字的示例可以写成：
    runBlocking {
        // 将一个整数区间转化为流
        (1..3).asFlow().collect { value -> println(value) }
    }
    //5. 过渡流操作符
    //可以使用操作符转换流，就像使用集合与序列一样。 过渡操作符应用于上游流，并返回下游流。
    // 这些操作符也是冷操作符，就像流一样。这类操作符本身不是挂起函数。它运行的速度很快，返回新的转换流的定义。
    //基础的操作符拥有相似的名字，比如 map 与 filter。 流与序列的主要区别在于这些操作符中的代码可以调用挂起函数。
    //举例来说，一个请求中的流可以使用 map 操作符映射出结果，即使执行一个长时间的请求操作也可以使用挂起函数来实现：
    suspend fun performRequest(request: Int): String {
        delay(1000) // 模仿长时间运行的异步工作
        return "response $request"
    }
    runBlocking {
        (1..3).asFlow() // 一个请求流
            .map { request -> performRequest(request) }
            .collect { response -> println(response) }
    }
    //5.1 转换操作符
    //在流转换操作符中，最通用的一种称为 transform。它可以用来模仿简单的转换，例如 map 与 filter，
    // 以及实施更复杂的转换。 使用 transform 操作符，我们可以 发射 任意值任意次。
    //比如说，使用 transform 我们可以在执行长时间运行的异步请求之前发射一个字符串并跟踪这个响应：
    runBlocking {
        (1..3).asFlow() // 一个请求流
            .transform { request ->
                emit("Making request $request")
                emit(performRequest(request))
            }
            .collect { response -> println(response) }
    }
    //5.2 限长操作符
    //限长过渡操作符（例如 take）在流触及相应限制的时候会将它的执行取消。协程中的取消操作总是通过抛出异常来执行，
    // 这样所有的资源管理函数（如 try {...} finally {...} 块）会在取消的情况下正常运行：
    fun numbers(): Flow<Int> = flow {
        try {
            emit(1)
            emit(2)
            println("This line will not execute")
            emit(3)
        } finally {
            println("Finally in numbers")
        }
    }
    runBlocking {
        numbers()
            .take(2) // 只获取前两个
            .collect { value -> println(value) }
    }

    //6. 末端流操作符
    //末端操作符是在流上用于启动流收集的挂起函数。 collect 是最基础的末端操作符，但是还有另外一些更方便使用的末端操作符：
    //a. 转化为各种集合，例如 toList 与 toSet。
    //b. 获取第一个（first）值与确保流发射单个（single）值的操作符。
    //c. 使用 reduce 与 fold 将流规约到单个值。
    runBlocking {
        val sum = (1..5).asFlow()
            .map { it * it } // 数字 1 至 5 的平方
            .reduce { a, b -> a + b } // 求和（末端操作符）
        println(sum)
    }
    //7. 流是连续的
    //流的每次单独收集都是按顺序执行的，除非进行特殊操作的操作符使用多个流。
    // 该收集过程直接在协程中运行，该协程调用末端操作符。
    // 默认情况下不启动新协程。
    // 从上游到下游每个过渡操作符都会处理每个发射出的值然后再交给末端操作符。
    //
    //请参见以下示例，该示例过滤偶数并将其映射到字符串：
    runBlocking<Unit> {
        (1..5).asFlow()
            .filter {
                println("Filter $it")
                it % 2 == 0
            }
            .map {
                println("Map $it")
                "string $it"
            }.collect {
                println("Collect $it")
            }
    }
    //8. 流上下文
    //流的收集总是在调用协程的上下文中发生。例如，如果有一个流 simple，然后以下代码在它的编写者指定的上下文中运行，而无论流 simple 的实现细节如何：
    fun log(msg: String) = println("[${Thread.currentThread().name}] $msg")
    fun simple7(): Flow<Int> = flow {
        log("Started simple flow")
        for (i in 1..3) {
            emit(i)
        }
    }
    runBlocking {
        withContext(Dispatchers.Default) {
            simple7().collect { value ->
                println(value) // 运行在指定上下文中
            }
        }
    }
    //流的该属性称为 上下文保存 。
    //所以默认的，flow { ... } 构建器中的代码运行在相应流的收集器提供的上下文中。
    // 举例来说，考虑打印线程的一个 simple 函数的实现， 它被调用并发射三个数字：

    runBlocking {
        simple7().collect { value -> log("Collected $value") }
    }
    //8.1 withContext 发出错误
    //然而，长时间运行的消耗 CPU 的代码也许需要在 Dispatchers.Default 上下文中执行，并且更新 UI 的代码也许需要
    // 在 Dispatchers.Main 中执行。通常，withContext 用于在 Kotlin 协程中改变代码的上下文，但是 flow {...}
    // 构建器中的代码必须遵循上下文保存属性，并且不允许从其他上下文中发射（emit）。
    //尝试运行下面的代码会产生异常：
    fun simple8(): Flow<Int> = flow {
        // 在流构建器中更改消耗 CPU 代码的上下文的错误方式
        withContext(Dispatchers.Default) {
            for (i in 1..3) {
                Thread.sleep(100) // 假装我们以消耗 CPU 的方式进行计算
                emit(i) // 发射下一个值
            }
        }
    }

    //    runBlocking {
//        simple8().collect { value -> println(value) }
//    }
    //8.2 flowOn 操作符
    //例外的是 flowOn 函数，该函数用于更改流发射的上下文。
    // 以下示例展示了更改流上下文的正确方法，该示例还通过打印相应线程的名字以展示它们的工作方式：
    fun simple9(): Flow<Int> = flow {
        for (i in 1..3) {
            Thread.sleep(100) // 假装我们以消耗 CPU 的方式进行计算
            log("Emitting $i")
            emit(i) // 发射下一个值
        }
    }.flowOn(Dispatchers.Default) // 在流构建器中改变消耗 CPU 代码上下文的正确方式
    runBlocking<Unit> {
        simple9().collect { value ->
            log("Collected $value")
        }
    }
    //注意，当收集发生在主线程中，flow { ... } 是如何在后台线程中工作的：
    //这里要观察的另一件事是 flowOn 操作符已改变流的默认顺序性。 现在收集发生在一个协程中（“coroutine#1”）
    // 而发射发生在运行于另一个线程中与收集协程并发运行的另一个协程（“coroutine#2”）中。
    // 当上游流必须改变其上下文中的 CoroutineDispatcher 的时候，flowOn 操作符创建了另一个协程。

    //9. 缓冲
    //从收集流所花费的时间来看，将流的不同部分运行在不同的协程中将会很有帮助，特别是当涉及到长时间运行的异步操作时。
    // 例如，考虑一种情况， 一个 simple 流的发射很慢，它每花费 100 毫秒才产生一个元素；而收集器也非常慢，
    // 需要花费 300 毫秒来处理元素。让我们看看从该流收集三个数字要花费多长时间：
    fun simple10(): Flow<Int> = flow {
        for (i in 1..3) {
            delay(100) // 假装我们异步等待了 100 毫秒
            emit(i) // 发射下一个值
        }
    }
    runBlocking<Unit> {
        val time = measureTimeMillis {
            simple10().collect { value ->
                delay(300) // 假装我们花费 300 毫秒来处理它
                println(value)
            }
        }
        println("Collected in $time ms")
    }
    //整个收集过程大约需要 1200 毫秒（3 个数字，每个花费 400 毫秒）
    //我们可以在流上使用 buffer 操作符来并发运行这个 simple 流中发射元素的代码以及收集的代码， 而不是顺序运行它们：
    runBlocking<Unit> {
        val time = measureTimeMillis {
            simple10()
                .buffer() // 缓冲发射项，无需等待
                .collect { value ->
                    delay(300) // 假装我们花费 300 毫秒来处理它
                    println(value)
                }
        }
        println("Collected in $time ms")
    }
    //它产生了相同的数字，只是更快了，由于我们高效地创建了处理流水线，
    // 仅仅需要等待第一个数字产生的 100 毫秒以及处理每个数字各需花费的 300 毫秒。这种方式大约花费了 1000 毫秒来运行。
    //注意，当必须更改 CoroutineDispatcher 时，flowOn 操作符使用了相同的缓冲机制， 但是我们在这里显式地请求缓冲而不改变执行上下文。
    //9.1 合并
    //当流代表部分操作结果或操作状态更新时，可能没有必要处理每个值，而是只处理最新的那个。
    // 在本示例中，当收集器处理它们太慢的时候， conflate 操作符可以用于跳过中间值。构建前面的示例：
    runBlocking<Unit> {
        val time = measureTimeMillis {
            simple10()
                .conflate() // 合并发射项，不对每个值进行处理
                .collect { value ->
                    println("Collecting $value")
                    delay(300) // 假装我们花费 300 毫秒来处理它
                    println("Done $value")
                }
        }
        println("Collected in $time ms")
    }
    //我们看到，虽然第一个数字仍在处理中，但第二个和第三个数字已经产生，因此第二个是 conflated ，只有最新的（第三个）被交付给收集器：
    //9.2 处理最新值
    //当发射器和收集器都很慢的时候，合并是加快处理速度的一种方式。它通过删除发射值来实现。
    // 另一种方式是取消缓慢的收集器，并在每次发射新值的时候重新启动它。
    // 有一组与 xxx 操作符执行相同基本逻辑的 xxxLatest 操作符，但是在新值产生的时候取消执行其块中的代码。
    // 让我们在先前的示例中尝试更换 conflate 为 collectLatest：
    runBlocking<Unit> {
        val time = measureTimeMillis {
            simple10()
                .collectLatest { value -> // 取消并重新发射最后一个值
                    println("Collecting $value")
                    delay(300) // 假装我们花费 300 毫秒来处理它
                    println("Done $value")
                }
        }
        println("Collected in $time ms")
    }
    //由于 collectLatest 的函数体需要花费 300 毫秒，但是新值每 100 秒发射一次，我们看到该代码块对每个值运行，但是只收集最后一个值

    //10. 组合多个流
    //10.1 zip
    //就像 Kotlin 标准库中的 Sequence.zip 扩展函数一样， 流拥有一个 zip 操作符用于组合两个流中的相关值：
    runBlocking<Unit> {
        val nums = (1..3).asFlow().onEach { delay(300) } // 数字 1..3
        val strs = flowOf("one", "two", "three").onEach { delay(400) }// 字符串
        val startTime = System.currentTimeMillis() // 记录开始的时间
        nums.zip(strs) { a, b -> "$a -> $b" } // 组合单个字符串
            .collect { println("$it at ${System.currentTimeMillis() - startTime} ms from start11111") } // 收集并打印
    }
    //10.2 Combine
    //当流表示一个变量或操作的最新值时，可能需要执行计算，这依赖于相应流的最新值，并且每当上游流产生值的时候都需要重新计算。
    // 这种相应的操作符家族称为 combine。
    //例如，先前示例中的数字如果每 300 毫秒更新一次，但字符串每 400 毫秒更新一次， 然后使用 zip 操作符合并它们，
    // 但仍会产生相同的结果， 尽管每 400 毫秒打印一次结果：
    //我们在本示例中使用 onEach 过渡操作符来延时每次元素发射并使该流更具说明性以及更简洁。
    runBlocking<Unit> {
        val nums = (1..3).asFlow().onEach { delay(300) } // 发射数字 1..3，间隔 300 毫秒
        val strs = flowOf("one", "two", "three").onEach { delay(400) } // 每 400 毫秒发射一次字符串
        val startTime = System.currentTimeMillis() // 记录开始的时间
        nums.zip(strs) { a, b -> "$a -> $b" } // 使用“zip”组合单个字符串
            .collect { value -> // 收集并打印
                println("$value at ${System.currentTimeMillis() - startTime} ms from start22222")
            }
    }
    //然而，当在这里使用 combine 操作符来替换 zip,我们得到了完全不同的输出，其中，nums 或 strs 流中的每次发射都会打印一行：
    runBlocking<Unit> {
        val nums = (1..3).asFlow().onEach { delay(300) } // 发射数字 1..3，间隔 300 毫秒
        val strs = flowOf("one", "two", "three").onEach { delay(400) } // 每 400 毫秒发射一次字符串
        val startTime = System.currentTimeMillis() // 记录开始的时间
        nums.combine(strs) { a, b -> "$a -> $b" } // 使用“combine”组合单个字符串
            .collect { value -> // 收集并打印
                println("$value at ${System.currentTimeMillis() - startTime} ms from start")
            }
    }

    //11. 展平流
    //流表示异步接收的值序列，所以很容易遇到这样的情况： 每个值都会触发对另一个值序列的请求。
    // 比如说，我们可以拥有下面这样一个返回间隔 500 毫秒的两个字符串流的函数：
    fun requestFlow(i: Int): Flow<String> = flow {
        emit("$i: First")
        delay(500) // 等待 500 毫秒
        emit("$i: Second")
    }
    //现在，如果我们有一个包含三个整数的流，并为每个整数调用 requestFlow，如下所示：
    (1..3).asFlow().map { requestFlow(it) }
    //然后我们得到了一个包含流的流（Flow<Flow<String>>），需要将其进行展平为单个流以进行下一步处理。
    // 集合与序列都拥有 flatten 与 flatMap 操作符来做这件事。
    // 然而，由于流具有异步的性质，因此需要不同的展平模式， 为此，存在一系列的流展平操作符。
    //11.1 flatMapConcat
    //连接模式由 flatMapConcat 与 flattenConcat 操作符实现。它们是相应序列操作符最相近的类似物。
    // 它们在等待内部流完成之前开始收集下一个值，如下面的示例所示：
    runBlocking {
        val startTime = System.currentTimeMillis() // 记录开始时间
        (1..3).asFlow().onEach { delay(100) } // 每 100 毫秒发射一个数字
            .flatMapConcat { requestFlow(it) }
            .collect { value -> // 收集并打印
                println("$value at ${System.currentTimeMillis() - startTime} ms from start aaaaaaaaa")
            }
    }
    //11.2 flatMapMerge
    //另一种展平模式是并发收集所有传入的流，并将它们的值合并到一个单独的流，以便尽快的发射值。
    // 它由 flatMapMerge 与 flattenMerge 操作符实现。他们都接收可选的用于限制并发收集的流的个数的
    // concurrency 参数（默认情况下，它等于 DEFAULT_CONCURRENCY）。
    runBlocking<Unit> {
        val startTime = System.currentTimeMillis() // 记录开始时间
        (1..3).asFlow().onEach { delay(100) } // 每 100 毫秒发射一个数字
            .flatMapMerge { requestFlow(it) }
            .collect { value -> // 收集并打印
                println("$value at ${System.currentTimeMillis() - startTime} ms from start bbbbbbbbb")
            }
    }
    //注意，flatMapMerge 会顺序调用代码块（本示例中的 { requestFlow(it) }），但是并发收集结果流，
    // 相当于执行顺序是首先执行 map { requestFlow(it) } 然后在其返回结果上调用 flattenMerge。
    //11.3 flatMapLatest
    //与 collectLatest 操作符类似（在"处理最新值" 小节中已经讨论过），也有相对应的“最新”展平模式，
    // 在发出新流后立即取消先前流的收集。 这由 flatMapLatest 操作符来实现。
    runBlocking<Unit> {
        val startTime = System.currentTimeMillis() // 记录开始时间
        (1..3).asFlow().onEach { delay(100) } // 每 100 毫秒发射一个数字
            .flatMapLatest { requestFlow(it) }
            .collect { value -> // 收集并打印
                println("$value at ${System.currentTimeMillis() - startTime} ms from start ccccccccc")
            }
    }
    //注意，flatMapLatest 在一个新值到来时取消了块中的所有代码 (本示例中的 { requestFlow(it) }）。
    // 这在该特定示例中不会有什么区别，由于调用 requestFlow 自身的速度是很快的，不会发生挂起， 所以不会被取消。
    // 然而，如果我们要在块中调用诸如 delay 之类的挂起函数，这将会被表现出来。

    //12. 流异常
    //当运算符中的发射器或代码抛出异常时，流收集可以带有异常的完成。 有几种处理异常的方法。
    //12.1  收集器 try 与 catch
    fun simple12(): Flow<Int> = flow {
        for (i in 1..3) {
            println("Emitting $i")
            emit(i) // 发射下一个值
        }
    }
    runBlocking<Unit> {
        try {
            simple12().collect { value ->
                println(value)
                check(value <= 1) { "Collected $value" }
            }
        } catch (e: Throwable) {
            println("Caught $e")
        }
    }
    //这段代码成功的在末端操作符 collect 中捕获了异常，并且， 如我们所见，在这之后不再发出任何值
    //12.2 一切都已捕获
    //前面的示例实际上捕获了在发射器或任何过渡或末端操作符中发生的任何异常。
    // 例如，让我们修改代码以便将发出的值映射为字符串， 但是相应的代码会产生一个异常：
    fun simple13(): Flow<String> =
        flow {
            for (i in 1..3) {
                println("Emitting $i")
                emit(i) // 发射下一个值
            }
        }
            .map { value ->
                check(value <= 1) { "Crashed on $value" }
                "string $value"
            }

    runBlocking<Unit> {
        try {
            simple13().collect { value -> println(value) }
        } catch (e: Throwable) {
            println("Caught $e")
        }
    }
    //13. 异常透明性
    //流必须对异常透明，即在 flow { ... } 构建器内部的 try/catch 块中发射值是违反异常透明性的。
    // 这样可以保证收集器抛出的一个异常能被像先前示例中那样的 try/catch 块捕获。
    //发射器可以使用 catch 操作符来保留此异常的透明性并允许封装它的异常处理。
    // catch 操作符的代码块可以分析异常并根据捕获到的异常以不同的方式对其做出反应：
    //可以使用 throw 重新抛出异常。
    //可以使用 catch 代码块中的 emit 将异常转换为值发射出去。
    //可以将异常忽略，或用日志打印，或使用一些其他代码处理它。
    fun simple14(): Flow<String> =
        flow {
            for (i in 1..3) {
                println("Emitting $i")
                emit(i) // 发射下一个值
            }
        }
            .map { value ->
                check(value <= 1) { "Crashed on $value" }
                "string $value"
            }
    runBlocking {
        simple14()
            .catch { e -> emit("Caught $e") } // 发射一个异常
            .collect { value -> println(value) }
    }
    //透明捕获
    //catch 过渡操作符遵循异常透明性，仅捕获上游异常（catch 操作符上游的异常，但是它下面的不是）。
    // 如果 collect { ... } 块（位于 catch 之下）抛出一个异常，那么异常会逃逸：
    fun simple15(): Flow<Int> = flow {
        for (i in 1..3) {
            println("Emitting $i")
            emit(i)
        }
    }
    runBlocking<Unit> {
        simple15()
            .catch { e -> println("Caught $e") } // 不会捕获下游异常
            .collect { value ->
                check(value <= 1) { "Collected $value" }
                println(value)
            }
    }
    //声明式捕获
    //我们可以将 catch 操作符的声明性与处理所有异常的期望相结合，将 collect 操作符的代码块移动到 onEach 中，并将其放到 catch 操作符之前
    runBlocking<Unit> {
        simple15()
            .onEach { value ->
                check(value <= 1) { "Collected $value" }
                println(value)
            }
            .catch { e -> println("Caught $e") }
            .collect()
    }
    //14. 流完成
    //当流收集完成时（普通情况或异常情况），它可能需要执行一个动作。
    // 你可能已经注意到，它可以通过两种方式完成：命令式或声明式。
    //命令式 finally 块：
    fun simple16(): Flow<Int> = (1..3).asFlow()
    runBlocking {
        try {
            simple16().collect { value -> println(value) }
        } finally {
            println("Done")
        }
    }
    //声明式处理
    //对于声明式，流拥有 onCompletion 过渡操作符，它在流完全收集时调用。
    runBlocking {
        simple16().onCompletion { println("Done") }
            .collect { value -> println(value) }
    }
    //onCompletion 的主要优点是其 lambda 表达式的可空参数 Throwable 可以用于确定流收集是正常完成
    // 还是有异常发生。在下面的示例中 simple 流在发射数字 1 之后抛出了一个异常：
    fun simple17(): Flow<Int> = flow {
        emit(1)
        throw RuntimeException()
    }
    runBlocking {
        simple17().onCompletion { cause -> if (cause != null) println("Flow completed exceptionally") }
            .catch { cause -> println("Caught exception") }
            .collect { value -> println(value) }
    }
}
